package cn.fan.vertd.utils;

import io.vertx.core.Vertx;
import io.vertx.core.json.JsonObject;
import io.vertx.kafka.client.consumer.KafkaConsumer;
import io.vertx.kafka.client.producer.KafkaProducer;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.HashMap;
import java.util.Map;

public class KafkaUtil {

    private final static Logger log = LogManager.getLogger();

    private KafkaConsumer<String, String> consumer;

    private KafkaProducer<String, String> producer;

    public KafkaUtil(Vertx vertx, JsonObject conf) {
        JsonObject kafkaProducer = conf.getJsonObject("kafkaProducer");
        Map<String, String> config = new HashMap<>();
        kafkaProducer.stream().iterator().forEachRemaining(entry -> {
            config.put(entry.getKey(), entry.getValue().toString());
        });
        this.producer = KafkaProducer.create(vertx, config);
    }

    public KafkaUtil(Vertx vertx, JsonObject conf, String groupId) {
        JsonObject kafkaProducer = conf.getJsonObject("kafka");
        Map<String, String> config = new HashMap<>();
        kafkaProducer.stream().iterator().forEachRemaining(entry -> {
            config.put(entry.getKey(), entry.getValue().toString());
        });
        JsonObject consumerGroupId = conf.getJsonObject("kafkaConsumerGroupId");
        config.put("group.id", consumerGroupId.getString(groupId));
        this.consumer = KafkaConsumer.create(vertx, config);
    }

    public KafkaConsumer<String, String> getConsumer() {
        return consumer;
    }

    public KafkaProducer<String, String> getProducer() {
        return producer;
    }
}
